package amqp_consumer

import (
	"context"
	_ "embed"
	"errors"
	"fmt"
	"lifi"
	"lifi/config"
	"lifi/plugins/common/tls"
	"lifi/plugins/sources"
	"log"
	"math/rand"
	"strings"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

type empty struct{}
type semaphore chan empty

// AMQPConsumer is the top level struct for this plugin
type AMQPConsumer struct {
	Brokers                []string          `yaml:"brokers"`
	Username               string            `yaml:"username"`
	Password               string            `yaml:"password"`
	Exchange               string            `yaml:"exchange"`
	ExchangeType           string            `yaml:"exchange_type"`
	ExchangeDurability     string            `yaml:"exchange_durability"`
	ExchangePassive        bool              `yaml:"exchange_passive"`
	ExchangeArguments      map[string]string `yaml:"exchange_arguments"`
	MaxUndeliveredMessages int               `yaml:"max_undelivered_messages"`

	// Queue Name
	Queue           string `yaml:"queue"`
	QueueDurability string `yaml:"queue_durability"`
	QueuePassive    bool   `yaml:"queue_passive"`

	// Binding Key
	BindingKey string `yaml:"binding_key"`

	// Controls how many messages the server will try to keep on the network
	// for consumers before receiving delivery acks.
	PrefetchCount int `yaml:"prefetch_count" json:"prefetch_count"`

	// AMQP Auth method
	AuthMethod string
	tls.ClientConfig

	ContentEncoding      string `yaml:"content_encoding"`
	MaxDecompressionSize int64  `yaml:"max_decompression_size"`

	deliveries map[lifi.TrackingID]amqp.Delivery

	conn   *amqp.Connection
	wg     *sync.WaitGroup
	cancel context.CancelFunc
}

func (a *AMQPConsumer) Close() error {
	//TODO implement me
	panic("implement me")
}

type externalAuth struct{}

func (a *externalAuth) Mechanism() string {
	return "EXTERNAL"
}
func (a *externalAuth) Response() string {
	return "\000"
}

func (a *AMQPConsumer) Init() error {
	// Defaults

	if len(a.Brokers) == 0 {
		a.Brokers = []string{"amqp://localhost:5672/influxdb"}
	}

	if a.AuthMethod == "" {
		a.AuthMethod = "PLAIN"
	}

	if a.ExchangeType == "" {
		a.ExchangeType = "topic"
	}

	if a.ExchangeDurability == "" {
		a.ExchangeDurability = "durable"
	}

	if a.QueueDurability == "" {
		a.QueueDurability = "durable"
	}

	if a.PrefetchCount == 0 {
		a.PrefetchCount = 50
	}

	if a.MaxUndeliveredMessages == 0 {
		a.MaxUndeliveredMessages = 1000
	}

	if a.MaxDecompressionSize <= 0 {
		a.MaxDecompressionSize = config.DefaultMaxDecompressionSize
	}

	return nil
}

func (a *AMQPConsumer) Gather(_ lifi.Collector) error {
	return nil
}

func (a *AMQPConsumer) createConfig() (*amqp.Config, error) {
	// make new tls config
	tlsCfg, err := a.ClientConfig.TLSConfig()
	if err != nil {
		return nil, err
	}

	var auth []amqp.Authentication
	if strings.ToUpper(a.AuthMethod) == "EXTERNAL" {
		auth = []amqp.Authentication{&externalAuth{}}
	} else if a.Username != "" || a.Password != "" {
		auth = []amqp.Authentication{
			&amqp.PlainAuth{
				Username: a.Username,
				Password: a.Password,
			},
		}
	}

	amqpConfig := amqp.Config{
		TLSClientConfig: tlsCfg,
		SASL:            auth, // if nil, it will be PLAIN
	}
	return &amqpConfig, nil
}

func (a *AMQPConsumer) Start(acc lifi.Collector) error {

	amqpConf, err := a.createConfig()
	if err != nil {
		return err
	}

	//a.decoder, err = internal.NewContentDecoder(a.ContentEncoding)
	if err != nil {
		return err
	}

	msgs, err := a.connect(amqpConf)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithCancel(context.Background())
	a.cancel = cancel

	a.wg = &sync.WaitGroup{}
	a.wg.Add(1)
	go func() {
		defer a.wg.Done()
		a.process(ctx, msgs, acc)
	}()

	go func() {
		for {
			err := <-a.conn.NotifyClose(make(chan *amqp.Error))
			if err == nil {
				break
			}

			log.Printf("Connection closed: %s; trying to reconnect", err)
			for {
				msgs, err := a.connect(amqpConf)
				if err != nil {
					log.Printf("AMQP connection failed: %s", err)
					time.Sleep(10 * time.Second)
					continue
				}

				a.wg.Add(1)
				go func() {
					defer a.wg.Done()
					a.process(ctx, msgs, acc)
				}()
				break
			}
		}
	}()

	return nil
}

func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, error) {
	brokers := a.Brokers

	p := rand.Perm(len(brokers))
	for _, n := range p {
		broker := brokers[n]
		log.Printf("Connecting to %q", broker)
		conn, err := amqp.DialConfig(broker, *amqpConf)
		if err == nil {
			a.conn = conn
			log.Printf("Connected to %q", broker)
			break
		}
		log.Printf("Error connecting to %q", broker)
	}

	if a.conn == nil {
		return nil, errors.New("could not connect to any broker")
	}

	ch, err := a.conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("failed to open a channel: %w", err)
	}

	if a.Exchange != "" {
		exchangeDurable := true
		if a.ExchangeDurability == "transient" {
			exchangeDurable = false
		}

		exchangeArgs := make(amqp.Table, len(a.ExchangeArguments))
		for k, v := range a.ExchangeArguments {
			exchangeArgs[k] = v
		}

		err = a.declareExchange(
			ch,
			exchangeDurable,
			exchangeArgs)
		if err != nil {
			return nil, err
		}
	}

	q, err := a.declareQueue(ch)
	if err != nil {
		return nil, err
	}

	if a.BindingKey != "" {
		err = ch.QueueBind(
			q.Name,       // queue
			a.BindingKey, // binding-key
			a.Exchange,   // exchange
			false,
			nil,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to bind a queue: %w", err)
		}
	}

	err = ch.Qos(
		a.PrefetchCount,
		0,     // prefetch-size
		false, // global
	)
	if err != nil {
		return nil, fmt.Errorf("failed to set QoS: %w", err)
	}

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // arguments
	)
	if err != nil {
		return nil, fmt.Errorf("failed establishing connection to queue: %w", err)
	}

	return msgs, err
}

func (a *AMQPConsumer) declareExchange(
	channel *amqp.Channel,
	exchangeDurable bool,
	exchangeArguments amqp.Table,
) error {
	var err error
	/*
		ExchangeDeclare：
		用于声明Exchange。如果Exchange不存在，则会创建一个新的Exchange；
		如果Exchange已经存在但参数不匹配，则会引发错误。
		如果Exchange已存在且参数匹配，则不会引发错误，也不会更改现有Exchange的任何属性。
		适用于创建新的Exchange或确保已存在的Exchange属性正确。

		ExchangeDeclarePassive：
		用于声明Exchange，但是如果Exchange不存在，则会引发错误。
		如果Exchange已存在，它将检查参数是否匹配，但不会创建新的Exchange。
		主要用于检查Exchange是否存在以及其属性是否匹配，而不会创建新的Exchange。

		总的来说，如果你希望确保Exchange存在并且具有特定的属性，你应该使用ExchangeDeclare。如果你只是想检查Exchange是否存在以及其属性是否匹配，而不想创建新的Exchange，那么你可以使用ExchangeDeclarePassive
	*/
	if a.ExchangePassive {
		err = channel.ExchangeDeclarePassive(
			a.Exchange,
			a.ExchangeType,
			exchangeDurable,
			false, // delete when unused
			false, // internal
			false, // no-wait
			exchangeArguments,
		)
	} else {
		err = channel.ExchangeDeclare(
			a.Exchange,
			a.ExchangeType,
			exchangeDurable,
			false, // delete when unused
			false, // internal
			false, // no-wait
			exchangeArguments,
		)
	}
	if err != nil {
		return fmt.Errorf("error declaring exchange: %w", err)
	}
	return nil
}

func (a *AMQPConsumer) declareQueue(channel *amqp.Channel) (*amqp.Queue, error) {
	var queue amqp.Queue
	var err error

	queueDurable := true
	if a.QueueDurability == "transient" {
		queueDurable = false
	}

	if a.QueuePassive {
		queue, err = channel.QueueDeclarePassive(
			a.Queue,      // queue
			queueDurable, // durable
			false,        // delete when unused
			false,        // exclusive
			false,        // no-wait
			nil,          // arguments
		)
	} else {
		queue, err = channel.QueueDeclare(
			a.Queue,      // queue
			queueDurable, // durable
			false,        // delete when unused
			false,        // exclusive
			false,        // no-wait
			nil,          // arguments
		)
	}
	if err != nil {
		return nil, fmt.Errorf("error declaring queue: %w", err)
	}
	return &queue, nil
}

// Read messages from queue and add them to the Accumulator
func (a *AMQPConsumer) process(ctx context.Context, msgs <-chan amqp.Delivery, ac lifi.Collector) {
	a.deliveries = make(map[lifi.TrackingID]amqp.Delivery)

	acc := ac.WithTracking(a.MaxUndeliveredMessages)
	sem := make(semaphore, a.MaxUndeliveredMessages)

	for {
		select {
		case <-ctx.Done():
			return
		case track := <-acc.Delivered():
			if a.AckDelivery(track) {
			}
			//收到消息
		case d, ok := <-msgs:
			if !ok {
				return
			}
			err := a.CollectMessage(acc, d)
			if err != nil {
				acc.AddError(err)
			}
			//收到空消息
		case sem <- empty{}:
			select {
			case <-ctx.Done():
				return
			case track := <-acc.Delivered():
				if a.AckDelivery(track) {
				}
			case d, ok := <-msgs:
				if !ok {
					return
				}
				err := a.CollectMessage(acc, d)
				if err != nil {
					// todo : add delivery info
					acc.AddError(err)
				}
			}
		}
	}
}

func (a *AMQPConsumer) CollectMessage(acc lifi.TrackingCollector, d amqp.Delivery) error {

	data := lifi.NewByteData(d.Body, "")
	id := acc.AddTrackingData(data)
	a.deliveries[id] = d
	return nil
}

func (a *AMQPConsumer) AckDelivery(track lifi.DeliveryInfo2) bool {
	delivery, ok := a.deliveries[track.ID()]
	if !ok {
		// Added by a previous connection
		return false
	}

	if track.Delivered() {
		err := delivery.Ack(true)
		//log.Printf("Ack delivery: %d", delivery.DeliveryTag)
		if err != nil {
			log.Printf("Unable to ack written delivery: %d: %v", delivery.DeliveryTag, err)
			a.conn.Close()
		}
	} else {
		err := delivery.Reject(false)
		if err != nil {
			log.Printf("Unable to reject failed delivery: %d: %v", delivery.DeliveryTag, err)
			a.conn.Close()
		}
	}

	delete(a.deliveries, track.ID())
	return true
}

func (a *AMQPConsumer) Stop() {
	// We did not connect successfully so there is nothing to do here.
	if a.conn == nil || a.conn.IsClosed() {
		return
	}
	a.cancel()
	a.wg.Wait()
	err := a.conn.Close()
	if err != nil && !errors.Is(err, amqp.ErrClosed) {
		log.Printf("Error closing AMQP connection: %s", err)
		return
	}
}

func init() {
	sources.Add("amqp_consumer", func() lifi.Source {
		return &AMQPConsumer{}
	})
}
